package com.classba.udf;


import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
import org.apache.hadoop.io.*;
import org.json.JSONException;
import org.json.JSONObject;

import java.util.LinkedHashMap;
import java.util.Map;


public class UDFFieldToString extends GenericUDF {

    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments)
            throws UDFArgumentException {

        if (arguments.length == 0) {
            throw new UDFArgumentLengthException("need arguments");
        }
        if (arguments.length % 2 == 1) {
            throw new UDFArgumentLengthException("need pair arguments");
        }

        int length = arguments.length;
        for (int i = 0; i < length; i = i + 2) {
            ObjectInspector key = arguments[i];

            if (!(key instanceof StringObjectInspector)) {
                throw new UDFArgumentLengthException("key must string");
            }
        }
        return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
    }

    @Override
    public Object evaluate(DeferredObject[] arguments) throws HiveException {

        JSONObject json = new JSONObject();
        int length = arguments.length;
        for (int i = 0; i < length; i = i + 2) {

            String key = arguments[i].get().toString();
            Object value = arguments[i + 1].get();

            try {

                if (value instanceof Text) {
                    json.put(key, value.toString());
                } else if (value instanceof IntWritable) {
                    json.put(key, ((IntWritable) value).get());
                } else if (value instanceof LongWritable) {
                    json.put(key, ((LongWritable) value).get());
                } else if (value instanceof DoubleWritable) {
                    json.put(key, ((DoubleWritable) value).get());
                } else if (value instanceof FloatWritable) {
                    json.put(key, ((FloatWritable) value).get());
                } else if (value instanceof NullWritable) {
                    json.put(key, "null");
                }

            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        return json.toString();
    }


    @Override
    public String getDisplayString(String[] children) {
        return "UDFFieldToString";
    }

}

